package net.bwie.realtime.jtp.log.function;

import net.bwie.realtime.jtp.log.bean.PageViewBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/19 15:39:23
 * @version: 1.0
 */
public class PageViewWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {

    private FastDateFormat dateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
    @Override
//    key：表示分组字段；window：表示时间窗口，可以获取窗口开始和结束时间；input：表示窗口中所有数据
// */
    // 1窗口时间
    public void apply(String key, TimeWindow window,
                      Iterable<PageViewBean> input, Collector<String> out) throws Exception {
//        窗口时间
        String windowStart = dateFormat.format(window.getStart());
        String windowEnd   = dateFormat.format(window.getEnd());


        Long pvCount = 0L;
        Long pvDuringTime = 0L;
        Long uvCount = 0L;
        Long sessionCount = 0L;


        for (PageViewBean bean : input){
            pvCount += bean.getPvCount();
            pvDuringTime += bean.getPvDuringTime();
            uvCount += bean.getUvCount();
            sessionCount += bean.getSessionCount();
        }

        // 3输出

        String output =  windowStart +"," + windowEnd +"," + key +"," + pvCount +"," + pvDuringTime +"," + uvCount +"," + sessionCount +","+ System.currentTimeMillis();


        out.collect(output);
    }
}
